package org.seed.rpc.clientstub;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.seed.rpc.clientstub.util.IDWorker;
import org.seed.rpc.common.codec.RpcDecoder;
import org.seed.rpc.common.codec.RpcEncoder;
import org.seed.rpc.common.io.RpcRequest;
import org.seed.rpc.common.io.RpcResponse;
import org.seed.rpc.core.RegistryClient;
import org.seed.rpc.core.RegistryOperator;
import org.seed.rpc.core.RuntimeLocal;
import org.seed.rpc.core.routing.Endpoint;
import org.seed.rpc.core.routing.RoutingStrategy;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 发送RPC服务请求
 *
 * <p>用法：RpcClient.newClientSend(request,ip,port);</p>
 */
@Slf4j
public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {


    private EventLoopGroup eventLoopGroup;
    private Bootstrap bootstrap;

    private String host;
    private int port;

    private RpcResponse response;

    /**
     * 客户端请求超时时间
     **/
    private Long requestTimeout = 8000L;

    private RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    private RpcClient(String host, int port, Long requestTimeout) {
        this.host = host;
        this.port = port;
        if (requestTimeout != null && requestTimeout.longValue() > 0l) {
            this.requestTimeout = requestTimeout;
        }
    }

    private RpcClient() {

    }

    private static ConcurrentLinkedQueue<RpcClient> clientCache = new ConcurrentLinkedQueue<RpcClient>();

    private static ExecutorService executor = Executors.newFixedThreadPool(8, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "rpc-client");
        }
    });

    /**
     * 数据初始化
     */ {
        // 创建并初始化 Netty 客户端 Bootstrap 对象
        eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(new RpcEncoder(RpcRequest.class)); // 编码 RPC 请求
                pipeline.addLast(new RpcDecoder(RpcResponse.class)); // 解码 RPC 响应
                pipeline.addLast(RpcClient.this); // 处理 RPC 响应
            }
        });
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
    }

    @Override
    public void channelRead0(ChannelHandlerContext handlerContext, RpcResponse response) throws Exception {
        this.response = response;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext handlerContext, Throwable cause) throws Exception {
        log.error("api caught exception: {}", cause.getMessage());
        cause.printStackTrace();
        handlerContext.close();
        throw new Exception(cause);
    }

    private RpcResponse send(RpcRequest request) throws Exception {
        try {
            // 连接 RPC 服务器
            ChannelFuture future = bootstrap.connect(host, port).sync();
            // 写入 RPC 请求数据并关闭连接
            Channel channel = future.channel();
            channel.writeAndFlush(request).sync();
            channel.closeFuture().sync();
            // 返回 RPC 响应对象
            return response;
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
    }


    /**
     * 发送rpc服务请求
     */
    public static RpcResponse newClientSend(String serviceName, RoutingStrategy routingStrategy, RpcRequest request, long requestTimeout) throws Exception {
        final Endpoint endpoint = routingStrategy.determineEndpoint(serviceName);
        RpcClient rpcClient = null;
        if (clientCache.isEmpty()) {
            rpcClient = new RpcClient(endpoint.getHost(), endpoint.getPort());
            triggerGenerateNewClients();
        } else {
            rpcClient = clientCache.poll();
            rpcClient.setHost(endpoint.getHost());
            rpcClient.setPort(endpoint.getPort());
        }
        final RpcClient finalRpcClient = rpcClient;
        Future<RpcResponse> future = executor.submit(new Callable<RpcResponse>() {
            @Override
            public RpcResponse call() {
                try {
                    return finalRpcClient.send(request);
                } catch (Exception e) {
                    log.error("netty-client request remote error: {}", e.getMessage());
                    if (e.getClass().getName().toLowerCase().endsWith("connectexception")) {
                        // 启动新线程, 检查连接状态
                        executor.execute(() -> {
                            checkLocalNetworkConnection(serviceName, endpoint);
                        });
                        // 网络连接失败, 重试一次
                        List<String> excepts = new ArrayList<>();
                        excepts.add(endpoint.getHost() + ":" + endpoint.getPort());
                        Endpoint endpointAnother = routingStrategy.determineEndpointExcept(serviceName, excepts);
                        RpcClient rpcClientAnother = new RpcClient(endpointAnother.getHost(), endpointAnother.getPort());
                        try {
                            return rpcClientAnother.send(request);
                        } catch (Exception exception) {
                            exception.printStackTrace();
                            return null;
                        }
                    }
                    e.printStackTrace();
                    return null;
                }
            }
        });
        try {
            return future.get(requestTimeout, TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            future.cancel(true);
            RpcResponse response = new RpcResponse();
            response.setRequestId(request.getRequestId());
            response.setException(new Exception("[" + request.getRequestId() + "] - method: " + request.getInterfaceName() + "." + request.getMethodName()));
            e.printStackTrace();
            return response;
        }
    }

    private static void triggerGenerateNewClients() {
        executor.execute(() -> {
            for (int i = 0; i < 20; i++) {
                clientCache.add(new RpcClient());
            }
        });
    }

    private static void checkLocalNetworkConnection(String serviceName, Endpoint endpoint) {
        // 检查本地网络连接状态
        // 1. 注册中心连接检查
        RegistryClient registryClient = RuntimeLocal.getRegistryClient();
        boolean registryConnection = RegistryOperator.testRegistryConnection(registryClient);
        if (!registryConnection) {
            return;
        }
        Map<String, List<String>> map = RegistryOperator.getServiceEndpointMap(registryClient);
        RuntimeLocal.cacheLocalServiceNameEndpointMap(map);
        // 2. 目标服务器连接检查
        final RpcRequest request = new RpcRequest();
        request.setRequestId(IDWorker.nextId62WithRand());
        request.setInterfaceName("org.seed.rpc.serverstub.health.RpcServiceHealth");
        request.setMethodName("ping");
        request.setParameterTypes(new Class[]{});
        request.setParameters(new Object[]{});
        RpcClient rpcClient = new RpcClient(endpoint.getHost(), endpoint.getPort());
        try {
            RpcResponse response = rpcClient.send(request);
            if (response != null && !response.hasException() && response.getResult().equals("PONG")) {
                return;
            }
        } catch (Exception e) {
        }
        RegistryOperator.unloadLocalEndpoint(serviceName, endpoint.getHost() + ":" + endpoint.getPort(), registryClient);
    }

    private String getHost() {
        return host;
    }

    private void setHost(String host) {
        this.host = host;
    }

    private int getPort() {
        return port;
    }

    private void setPort(int port) {
        this.port = port;
    }


}
